package org.alanlau;

import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Component;

/**
 * 消费者
 * @author alan
 * @date  2021/4/10 12:25
 * @version 1.0
 */
@Component
public class MyConsumer {
    /**
     * @KafkaListener(groupId = "testGroup", topicPartitions = {
     *             @TopicPartition(topic = "topic1", partitions = {"0", "1"}),
     *             @TopicPartition(topic = "topic2", partitions = "0",
     *                     partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "100"))
     *     },concurrency = "6")
     *     // concurrency就是同组下的消费者个数，就是并发消费数，必须小于等于分区总数
     * @param record 消息记录
     * @param ack 签收
     */
    @KafkaListener(topics = "my-replicated-topic", groupId = "testGroup", topicPartitions = {
            @TopicPartition(topic = "my-replicated-topic", partitions = {"0", "1"})
    })
    public void listenTestGroup(ConsumerRecord<String, String> record, Acknowledgment ack) {
        System.out.println(String.format("listener1 --> topic: {%s}, partition: {%d}, offset: {%d}, key: {%s}, value: {%s}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value()));
        //手动提交offset
        ack.acknowledge();
    }

    @KafkaListener(topics = "my-replicated-topic", groupId = "testGroup", topicPartitions = {
            @TopicPartition(topic = "my-replicated-topic", partitions = {"2"})
    })
    public void listenTestGroup2(ConsumerRecord<String, String> record, Acknowledgment ack) {
        System.out.println(String.format("listener2 --> topic: {%s}, partition: {%d}, offset: {%d}, key: {%s}, value: {%s}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value()));
        //手动提交offset
        //ack.acknowledge();
    }

    /**
     * 配置多个消费组
     * @param record 消息
     * @param ack 签收
     */
    @KafkaListener(groupId = "testGroup2", topics = {TopicConstants.TOPIC_NAME, TopicConstants.TOPIC_NAME2})
    public void listenTest2Group(ConsumerRecord<String, String> record, Acknowledgment ack) {
        System.out.println(String.format("listener3 --> topic: {%s}, partition: {%d}, offset: {%d}, key: {%s}, value: {%s}",
                record.topic(), record.partition(), record.offset(), record.key(), record.value()));
        ack.acknowledge();
    }
}
